package com.ctl.hadoop;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URISyntaxException;

public class Test {
    static {
        System.setProperty("HADOOP_USER_NAME", "root"); //执行在hsdf服务器用户(hdfs文件路径用户)
    }

    static String hdfsUrl = "hdfs://master:9000";

    /**
     * mkdir("/20200817");
     *
     * @param path
     * @throws IOException
     */
    public static void mkdir(String path) throws IOException {
        //读取配置文件
        Configuration conf = new Configuration();
        //获取文件系统
        FileSystem fs = FileSystem.get(URI.create(hdfsUrl), conf);

        Path srcPath = new Path(path);
        //调用mkdir（）创建目录，（可以一次性创建，以及不存在的父目录）
        boolean flag = fs.mkdirs(srcPath);
        if (flag) {
            System.out.println("create dir ok!");
        } else {
            System.out.println("create dir failure");
        }

        //关闭文件系统
        fs.close();
    }

    /**
     * 删除文件或者文件目录
     *
     * @param filePath
     * @throws IOException
     */
    public static void rmdir(String filePath) throws IOException {
        //读取配置文件
        Configuration conf = new Configuration();
        //获取文件系统
        FileSystem fs = FileSystem.get(URI.create(hdfsUrl), conf);
        Path path = new Path(filePath);

        //调用deleteOnExit(）
        boolean flag = fs.deleteOnExit(path);
        //	fs.delete(path);
        if (flag) {
            System.out.println("delete ok!");
        } else {
            System.out.println("delete failure");
        }

        //关闭文件系统
        fs.close();
    }

    /**
     * 创建文件
     **/
    public static void createFile(String dst, byte[] contents) throws IOException {
        //读取配置文件
        Configuration conf = new Configuration();
        //获取文件系统
        FileSystem fs = FileSystem.get(URI.create(hdfsUrl), conf);
        //目标路径
        Path dstPath = new Path(dst);
        //打开一个输出流
        FSDataOutputStream outputStream = fs.create(dstPath);
        outputStream.write(contents);

        //关闭文件系统
        outputStream.close();
        fs.close();
        System.out.println("文件创建成功！");

    }

    /**
     * 列出文件
     **/
    public static void listFile(String path) throws IOException {
        //读取配置文件
        Configuration conf = new Configuration();
        //获取文件系统
        FileSystem fs = FileSystem.get(URI.create(hdfsUrl), conf);
        //获取文件或目录状态
        FileStatus[] fileStatus = fs.listStatus(new Path(path));
        //打印文件的路径
        for (FileStatus file : fileStatus) {
            System.out.println(file.getPath());
        }

        //关闭文件系统
        fs.close();
    }

    /**
     * 上传本地文件
     **/
    public static void uploadFile(String src, String dst) throws IOException {
        //读取配置文件
        Configuration conf = new Configuration();
        //获取文件系统
        FileSystem fs = FileSystem.get(URI.create(hdfsUrl), conf);
        Path srcPath = new Path(src); //原路径
        Path dstPath = new Path(dst); //目标路径
        //调用文件系统的文件复制函数,前面参数是指是否删除原文件，true为删除，默认为false
        fs.copyFromLocalFile(false, srcPath, dstPath);

        //打印文件路径
        System.out.println("Upload to " + conf.get("fs.default.name"));
        System.out.println("------------list files------------" + "\n");
        FileStatus[] fileStatus = fs.listStatus(dstPath);
        for (FileStatus file : fileStatus) {
            System.out.println(file.getPath());
        }

        //关闭文件系统
        fs.close();
    }

    /**
     * 文件重命名
     **/
    public static void renameFile(String oldName, String newName) throws IOException {
        //读取配置文件
        Configuration conf = new Configuration();
        //获取文件系统
        FileSystem fs = FileSystem.get(URI.create(hdfsUrl), conf);
        Path oldPath = new Path(oldName);
        Path newPath = new Path(newName);

        boolean flag = fs.rename(oldPath, newPath);
        if (flag) {
            System.out.println("rename ok!");
        } else {
            System.out.println("rename failure");
        }

        //关闭文件系统
        fs.close();
    }

    /**
     * 读取文件内容
     *
     * @param uri
     * @throws IOException
     */
    public static void readFile(String uri) throws IOException, URISyntaxException {
        //读取配置文件
        Configuration conf = new Configuration();
        //获取文件系统
        FileSystem fs = null;
        fs = FileSystem.get(URI.create(hdfsUrl), conf);

        InputStream in = null;
        try {
            in = fs.open(new Path(uri));
            //复制到标准输出流
            IOUtils.copyBytes(in, System.out, 4096, false);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            IOUtils.closeStream(in);
        }
    }

    /**
     * 判断目录是否存在
     */

    public static boolean existDir(String filePath, boolean create) {
        boolean flag = false;
        //判断是否存在
        if (StringUtils.isEmpty(filePath)) {
            return flag;
        }

        Path path = new Path(filePath);
        //读取配置文件
        Configuration conf = new Configuration();
        try {
            //获取文件系统
            FileSystem fs = FileSystem.get(URI.create(hdfsUrl), conf);

            //或者create为true
            if (create) {
                //如果文件不存在
                if (!fs.exists(path)) {
                    fs.mkdirs(path);
                }
            }

            //判断是否为目录
            if (fs.isDirectory(path)) {
                flag = true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return flag;
    }

    /**
     * 添加到文件的末尾(src为本地地址，dst为hdfs文件地址)
     *
     * @throws IOException
     */
    public static void appendFile(String src, String dst) throws IOException {
        //读取配置文件
        Configuration conf = new Configuration();
        //获取文件系统
        FileSystem fs = FileSystem.get(URI.create(hdfsUrl), conf);
        Path dstPath = new Path(dst);
        //创建需要写入的文件流
        InputStream in = new BufferedInputStream(new FileInputStream(src));

        //文件输出流写入
        FSDataOutputStream out = fs.append(dstPath);
        IOUtils.copyBytes(in, out, in.available(), true);
        fs.close();
    }

    public static void main(String[] args) throws IOException, URISyntaxException {
        String path = "/20200817";
        mkdir(path);
        createFile(path + "/123.txt", "123".getBytes());
        uploadFile("/home/software/note1.txt", path + "/note1.txt");
        renameFile(path + "/note1.txt", path + "/note11.txt");
        listFile(path);
        renameFile(path + "/note11.txt", path + "/note1.txt");
        listFile(path);
        readFile(path + "/note1.txt");
        existDir(path, false);
        appendFile("/home/software/note1.txt", path + "/note1.txt");
        readFile(path + "/note1.txt");
        //rmdir("/20200817");
    }
}
